package com.atguigu.gmall.realtime.common.base;

import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.SQLUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public abstract class BaseSQLApp {
    public void start(int port, int parallelism, String ck){
        // TODO 1.获取环境
        Configuration configuration = new Configuration();
        configuration.set(RestOptions.PORT,port);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
        env.setParallelism(parallelism);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        // TODO 2.设置检查点
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        /*
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        checkpointConfig.setCheckpointTimeout(60000L);
        checkpointConfig.setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        checkpointConfig.setMaxConcurrentCheckpoints(1);
        checkpointConfig.setMinPauseBetweenCheckpoints(2000L);
        env.setStateBackend(new HashMapStateBackend());//储存在taskManager堆内存
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 3000L));
        checkpointConfig.setCheckpointStorage("hdfs://hadoop102:8020/ck" + ck);
        System.setProperty("HADOOP_USER_NAME", "atguigu");
        */
        // TODO 3.业务处理
        handle(env , tableEnv);
    }

    public abstract  void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv);

    protected void readOdsDb(StreamTableEnvironment tableEnv , String groupId) {
        // TODO 3.从kafka的topic_db主题获取数据，并创建动态表---kafka连接器
        tableEnv.executeSql("CREATE TABLE topic_db (\n" +
                "  `database` STRING,\n" +
                "  `table` STRING,\n" +
                "  `type` STRING,\n" +
                "  `data` MAP<STRING, STRING>,\n" +
                "  `old` MAP<STRING, STRING>,\n" +
                "   ts BIGINT,\n" +
                "   pt AS PROCTIME(),\n" +
                "   et AS TO_TIMESTAMP_LTZ(ts, 0),\n" +
                "   WATERMARK FOR et AS et\n" +
                ")" + SQLUtil.getKafkaDDL(Constant.TOPIC_DB, groupId));
    }

    protected void readBaseDic(StreamTableEnvironment tableEnv) {
        // TODO 从Hbase获取dic信息，并创建动态表---Hbase连接器
        tableEnv.executeSql("CREATE TABLE base_dic (\n" +
                "    dic_code STRING,\n" +
                "    info ROW<dic_name STRING>,\n" +
                "    PRIMARY KEY (dic_code) NOT ENFORCED\n" +
                ") " + SQLUtil.getHbaseDDL("dim_base_dic"));
    }
}
